package com.fubang.rxjavalearning.mergeOperator;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;

import com.fubang.rxjavalearning.R;
import com.fubang.rxjavalearning.mapOperator.House;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Observer;
import rx.functions.Func1;
import rx.functions.Func2;

public class MergeActivity extends AppCompatActivity {

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_merge);
        //merge
        final String[] letters = {"A", "B", "C", "D", "E", "F", "G", "H"};
        Observable<String> observable1 = Observable.interval(300, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, String>() {
                    @Override
                    public String call(Long aLong) {
                        return letters[aLong.intValue()];
                    }
                }).take(letters.length);

        Observable<Long> observable2 = Observable.interval(500, TimeUnit.MILLISECONDS)
                .take(5);
//
//        Observable.merge(observable1, observable2)
//                .subscribe(new Observer<Serializable>() {
//                    @Override
//                    public void onCompleted() {
//                        System.exit(0);
//                    }
//
//                    @Override
//                    public void onError(Throwable e) {
//                        System.out.println("error");
//                    }
//
//                    @Override
//                    public void onNext(Serializable serializable) {
//                        System.out.println(serializable.toString());
//                    }
//                });
        System.out.println("----------------------------------------");
        //concat
//        Observable.concat(observable1, observable2)
//                .subscribe(new Observer<Serializable>() {
//                    @Override
//                    public void onCompleted() {
//                        System.exit(0);
//                    }
//
//                    @Override
//                    public void onError(Throwable e) {
//                        System.out.println("error");
//                    }
//
//                    @Override
//                    public void onNext(Serializable serializable) {
//                        System.out.println(serializable.toString());
//                    }
//                });

        System.out.println("----------------------------------------");
        //zip
        Observable.zip(observable1, observable2, new Func2<String, Long, String>() {
            @Override
            public String call(String s, Long aLong) {
                return s + aLong;
            }
        }).subscribe(new Observer<Serializable>() {
                    @Override
                    public void onCompleted() {
                        System.exit(0);
                    }

                    @Override
                    public void onError(Throwable e) {
                        System.out.println("error");
                    }

                    @Override
                    public void onNext(Serializable serializable) {
                        System.out.println(serializable.toString());
                    }
                });

        System.out.println("----------------------------------------");
        final ArrayList<House> houses = new ArrayList<>();
        houses.add(new House("中粮·海景壹号", "中粮海景壹号新出大平层！总价4500W起"));
        houses.add(new House("竹园新村", "满五唯一，黄金地段"));
        houses.add(new House("中粮·海景壹号", "毗邻汤臣一品"));
        houses.add(new House("竹园新村", "顶层户型，两室一厅"));
        houses.add(new House("中粮·海景壹号", "南北通透，豪华五房"));

        Observable<House> houseObservable = Observable.interval(1, TimeUnit.MILLISECONDS)
                .map(new Func1<Long, House>() {
                    @Override
                    public House call(Long aLong) {
                        return houses.get(aLong.intValue());
                    }
                }).take(houses.size());

        Observable<Long> longObservable = Observable.interval(1, TimeUnit.MILLISECONDS);
        houseObservable.join(longObservable,
                new Func1<House, Observable<Long>>() {
                    @Override
                    public Observable<Long> call(House house) {
                        return Observable.timer(2, TimeUnit.SECONDS);
                    }
                },
                new Func1<Long, Observable<Long>>() {
                    @Override
                    public Observable<Long> call(Long aLong) {
                        return Observable.timer(0, TimeUnit.SECONDS);
                    }
                },
                new Func2<House, Long, String>() {
                    @Override
                    public String call(House house, Long aLong) {
                        return aLong + "-->" + house.getDesc();
                    }
                }).subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {
                System.exit(0);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error:"+e.getMessage());
            }

            @Override
            public void onNext(String s) {
                System.out.println(s);
            }
        });

    }
}
